package org.jetlinks.pro.things.impl.rpc;

import lombok.AllArgsConstructor;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.ThingRpcSupport;
import org.jetlinks.core.things.ThingType;
import org.jetlinks.core.things.ThingsRegistry;
import org.jetlinks.pro.things.ThingConstants;
import org.jetlinks.pro.things.ThingMessageConverter;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.function.Function;



/**
 * 基本的物的RPC支持
 *
 * @author zhouhao
 * @since 1.12
 */
@AllArgsConstructor
public class SimpleThingsRpcSupport implements ThingRpcSupport {

    //物注册中心,统一管理物的基础信息以及配置等信息
    private final ThingsRegistry registry;

    //物类型定义
    private final ThingType thingType;

    //物ID
    private final String thingId;

    //物消息转换器,将消息模版和上下文转为新的消息
    private final ThingMessageConverter messageConverter;

    /**
     * 将请求消息转换为对应物的连接器支持
     *
     * @param message 请求消息
     * @return 物连接器
     */
    public Flux<? extends ThingMessage> call(ThingMessage message) {
        boolean useSourceThing = StringUtils.isEmpty(this.thingId);

        //如果基本物RPC支持中定义的物ID为空，则使用当前消息的物ID
        String thingId = useSourceThing ? message.getThingId():this.thingId;

        return Mono
            .zip(
                //从注册中心拿到物实例接口
                registry.getThing(thingType, thingId),
                //转换请求消息，如串行执行时，将上一个节点的返回结果当做上下文，用于转换当前请求消息
                this.convert(message)
                    .doOnNext(newMsg->{
                        //如果基本物RPC支持中定义的物ID不为空，则使用定义的物ID作为消息物ID。避免调用其它物RPC时，调用结果消息中物ID错误问题
                        if(useSourceThing){
                            newMsg.thingId(thingType,message.getThingId());
                        }
                    }),
                (device, msg) -> device.rpc().call(msg)
            )
            .flatMapMany(Function.identity());
    }

    /**
     * 使用物消息header中的上下文来转换请求消息 {@link ThingConstants#contextVar}
     *
     * @param message 请求消息
     */
    protected Mono<ThingMessage> convert(ThingMessage message) {

        return messageConverter
            .convert(message.getHeaderOrDefault(ThingConstants.contextVar), message);
    }
}
